package com.suxg.hbase.utils;

import java.io.IOException;
import java.nio.ByteBuffer;
import java.util.List;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.HBaseConfiguration;
import org.apache.hadoop.hbase.HColumnDescriptor;
import org.apache.hadoop.hbase.HTableDescriptor;
import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.client.Connection;
import org.apache.hadoop.hbase.client.ConnectionFactory;
import org.apache.hadoop.hbase.client.Delete;
import org.apache.hadoop.hbase.client.Get;
import org.apache.hadoop.hbase.client.HBaseAdmin;
import org.apache.hadoop.hbase.client.Put;
import org.apache.hadoop.hbase.client.Result;
import org.apache.hadoop.hbase.client.ResultScanner;
import org.apache.hadoop.hbase.client.Scan;
import org.apache.hadoop.hbase.client.Table;
import org.apache.hadoop.hbase.filter.FilterList;
import org.apache.hadoop.hbase.util.Bytes;
import org.apache.hadoop.hbase.util.Pair;

public class HBaseUtils {

	private static Connection connection;

	static {
		Configuration configuration = HBaseConfiguration.create();
		configuration.set("hbase.zookeeper.property.clientPort", "2181");
		// 如果是集群 则主机名用逗号分隔
		configuration.set("hbase.zookeeper.quorum", "localhost:2181");
		configuration.set("hbase.nodeParent", "localhost:2181");

		try {
			connection = ConnectionFactory.createConnection(configuration);
		} catch (IOException e) {
			e.printStackTrace();
		}
	}

	/**
	 * 创建 HBase 表
	 * 
	 * @param tableName      表名
	 * @param columnFamilies 列族的数组
	 */
	public static boolean createTable(String tableName, List<String> columnFamilies) {
		try {
			HBaseAdmin admin = (HBaseAdmin) connection.getAdmin();
			if (admin.tableExists(tableName)) {
				return false;
			}
			HTableDescriptor tableDescriptor = new HTableDescriptor(TableName.valueOf(tableName));
			columnFamilies.forEach(columnFamily -> {
				HColumnDescriptor columnDescriptor = new HColumnDescriptor(columnFamily);
				columnDescriptor.setMaxVersions(1);
				tableDescriptor.addFamily(columnDescriptor);
			});
			admin.createTable(tableDescriptor);
		} catch (IOException e) {
			e.printStackTrace();
		}
		return true;
	}

	/**
	 * 删除 hBase 表
	 *
	 * @param tableName 表名
	 */
	public static boolean deleteTable(String tableName) {
		try {
			HBaseAdmin admin = (HBaseAdmin) connection.getAdmin();
			// 删除表前需要先禁用表
			admin.disableTable(tableName);
			admin.deleteTable(tableName);
		} catch (Exception e) {
			e.printStackTrace();
		}
		return true;
	}

	/**
	 * 插入数据
	 *
	 * @param tableName        表名
	 * @param rowKey           唯一标识
	 * @param columnFamilyName 列族名
	 * @param qualifier        列标识
	 * @param value            数据
	 */
	public static boolean putRow(String tableName, String rowKey, String columnFamilyName, String qualifier,
			String value) {
		try {
			Table table = connection.getTable(TableName.valueOf(tableName));
			Put put = new Put(Bytes.toBytes(rowKey));
			put.addColumn(Bytes.toBytes(columnFamilyName), Bytes.toBytes(qualifier), Bytes.toBytes(value));
			table.put(put);
			table.close();
		} catch (IOException e) {
			e.printStackTrace();
		}
		return true;
	}

	/**
	 * 根据 rowKey 获取指定行的数据
	 *
	 * @param tableName 表名
	 * @param rowKey    唯一标识
	 */
	public static Result getRow(String tableName, String rowKey) {
		try {
			Table table = connection.getTable(TableName.valueOf(tableName));
			Get get = new Get(Bytes.toBytes(rowKey));
			return table.get(get);
		} catch (IOException e) {
			e.printStackTrace();
		}
		return null;
	}

	/**
	 * 获取指定行指定列 (cell) 的最新版本的数据
	 *
	 * @param tableName    表名
	 * @param rowKey       唯一标识
	 * @param columnFamily 列族
	 * @param qualifier    列标识
	 */
	public static String getCell(String tableName, String rowKey, String columnFamily, String qualifier) {
		try {
			Table table = connection.getTable(TableName.valueOf(tableName));
			Get get = new Get(Bytes.toBytes(rowKey));
			if (!get.isCheckExistenceOnly()) {
				get.addColumn(Bytes.toBytes(columnFamily), Bytes.toBytes(qualifier));
				Result result = table.get(get);
				byte[] resultValue = result.getValue(Bytes.toBytes(columnFamily), Bytes.toBytes(qualifier));
				return Bytes.toString(resultValue);
			} else {
				return null;
			}

		} catch (IOException e) {
			e.printStackTrace();
		}
		return null;
	}

	/**
	 * 检索全表
	 *
	 * @param tableName 表名
	 */
	public static ResultScanner getScanner(String tableName) {
		try {
			Table table = connection.getTable(TableName.valueOf(tableName));
			Scan scan = new Scan();
			// 进行全表扫描加开始，结束行，优化查询效率
//            scan.setStartRow(null);
//            scan.setStopRow(null);
			return table.getScanner(scan);
		} catch (IOException e) {
			e.printStackTrace();
		}
		return null;
	}

	/**
	 * 检索表中指定数据
	 *
	 * @param tableName  表名
	 * @param filterList 过滤器
	 */

	public static ResultScanner getScanner(String tableName, FilterList filterList) {
		try {
			Table table = connection.getTable(TableName.valueOf(tableName));
			Scan scan = new Scan();
			scan.setFilter(filterList);
			return table.getScanner(scan);
		} catch (IOException e) {
			e.printStackTrace();
		}
		return null;
	}

	/**
	 * 检索表中指定数据
	 *
	 * @param tableName   表名
	 * @param startRowKey 起始 RowKey
	 * @param endRowKey   终止 RowKey
	 * @param filterList  过滤器
	 */

	public static ResultScanner getScanner(String tableName, String startRowKey, String endRowKey,
			FilterList filterList) {
		try {
			Table table = connection.getTable(TableName.valueOf(tableName));
			Scan scan = new Scan();
			scan.setStartRow(Bytes.toBytes(startRowKey));
			scan.setStopRow(Bytes.toBytes(endRowKey));
			scan.setFilter(filterList);
			return table.getScanner(scan);
		} catch (IOException e) {
			e.printStackTrace();
		}
		return null;
	}

	/**
	 * 删除指定行记录
	 *
	 * @param tableName 表名
	 * @param rowKey    唯一标识
	 */
	public static boolean deleteRow(String tableName, String rowKey) {
		try {
			Table table = connection.getTable(TableName.valueOf(tableName));
			Delete delete = new Delete(Bytes.toBytes(rowKey));
			table.delete(delete);
		} catch (IOException e) {
			e.printStackTrace();
		}
		return true;
	}

	/**
	 * 删除指定行的指定列
	 *
	 * @param tableName  表名
	 * @param rowKey     唯一标识
	 * @param familyName 列族
	 * @param qualifier  列标识
	 */
	public static boolean deleteColumn(String tableName, String rowKey, String familyName, String qualifier) {
		try {
			Table table = connection.getTable(TableName.valueOf(tableName));
			Delete delete = new Delete(Bytes.toBytes(rowKey));
			delete.addColumn(Bytes.toBytes(familyName), Bytes.toBytes(qualifier));
			table.delete(delete);
			table.close();
		} catch (IOException e) {
			e.printStackTrace();
		}
		return true;
	}

	/**
	 * 插入数据
	 *
	 * @param tableName        表名
	 * @param rowKey           唯一标识
	 * @param columnFamilyName 列族名
	 * @param pairList         列标识和值的集合
	 */
	public static boolean putRow(String tableName, String rowKey, String columnFamilyName,
			List<Pair<String, String>> pairList) {
		try {
			Table table = connection.getTable(TableName.valueOf(tableName));
			Put put = new Put(Bytes.toBytes(rowKey));
//            pairList.forEach(pair -> put.addColumn(Bytes.toBytes(columnFamilyName), Bytes.toBytes(pair.getKey()), Bytes.toBytes(pair.getValue())));

			for (Pair<String, String> pair : pairList) {
				put.addColumn(Bytes.toBytes(columnFamilyName), pair.getFirst().getBytes(), pair.getSecond().getBytes());
			}
			table.put(put);
			table.close();
		} catch (IOException e) {
			e.printStackTrace();
		}
		return true;

	}

}
